package com.sunday.common.mq.kafka.study.spring.e16_Handling_Exceptions;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.ContainerCustomizer;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.CommonErrorHandler;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

import java.util.Map;

;

/**
 * https://docs.spring.io/spring-kafka/docs/current/reference/html/#manual-acknowledgment
 * 手动确认
 */
@Slf4j
@Configuration
public class ManualAckListenerContainerConfiguration {

    /**
     * 这些是Spring Kafka中AckMode枚举类型的不同取值，用于指定消息确认模式。下面是对每个取值的解释：
     * <p>
     * RECORD: 每处理一条消息后立即提交该消息的偏移量（offset）。
     * BATCH: 在所有从上一个轮询（poll）返回的消息都被处理后，批量提交它们的偏移量。
     * TIME: 在超过指定的时间间隔（通过ContainerProperties.setAckTime(long)设置）后，提交待确认的偏移量。
     * COUNT: 在达到指定的记录数（通过ContainerProperties.setAckCount(int)设置）后，提交待确认的偏移量。
     * COUNT_TIME: 在达到指定的记录数或超过指定的时间间隔后，提交待确认的偏移量。
     * MANUAL: 监听器负责手动确认消息接收。需要使用AcknowledgingMessageListener接口，当所有从上一个轮询返回的消息都被监听器处理后，才会提交偏移量。
     * MANUAL_IMMEDIATE: 监听器负责手动确认消息接收。需要使用AcknowledgingMessageListener接口，如果在调用消费者线程上的Acknowledgment进行确认，则立即执行提交；否则，在所有从上一个轮询返回的消息都被监听器处理后，才会提交偏移量。注意：如果有时在调用线程上进行确认，有时不进行确认，则结果是不确定的。
     * 这些确认模式用于控制何时以及如何提交消费者的偏移量，以确保消息被成功处理和提交。您可以根据实际需求选择适合的确认模式，并将其配置到@KafkaListener注解或相关的Kafka监听容器中。
     * <p>
     * <p>
     * MANUAL 和 MANUAL_IMMEDIATE 是 Spring Kafka 中的消费者确认模式，用于标识监听器负责手动确认消息接收的方式。它们的区别如下：
     * MANUAL 模式：在 MANUAL 模式下，需要使用 AcknowledgingMessageListener 接口来实现消费者逻辑，并在处理完消息后调用 Acknowledgment 对象的 acknowledge() 方法进行手动确认。在这种模式下，偏移量的提交是在所有从上一个轮询返回的消息都被监听器处理完毕后才进行的。
     * MANUAL_IMMEDIATE 模式：在 MANUAL_IMMEDIATE 模式下，同样需要使用 AcknowledgingMessageListener 接口来实现消费者逻辑，并在处理完消息后调用 Acknowledgment 对象的 acknowledge() 方法进行手动确认。不同之处在于，如果确认操作发生在调用消费者线程上的 Acknowledgment 对象时，偏移量将立即提交。如果确认操作没有在调用线程上执行，则会将确认操作排队，并在所有从上一个轮询返回的消息都被监听器处理完毕后才提交偏移量。需要注意的是，如果有时在调用线程上进行确认，有时不进行确认，则结果是不确定的。
     * 总结而言，MANUAL 模式要求在所有消息被处理后统一提交偏移量，而 MANUAL_IMMEDIATE 模式允许根据确认操作的发生时间来灵活提交偏移量，使得能够立即提交或在消息处理完毕后再提交。选择哪种模式取决于您对消息处理和偏移量提交时机的需求。
     */
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaManualAckListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory,
            ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer,
            KafkaProperties properties
    ) {

        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory
                .getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties())));
        kafkaContainerCustomizer.ifAvailable(factory::setContainerCustomizer);
//        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        /**
         * 手动两个值都需要设置
         * Caused by: java.lang.IllegalStateException: Consumer cannot be configured for auto commit for ackMode MANUAL
         */
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getConsumerFactory().updateConfigs(Map.of(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false));

        return factory;
    }

}
